package com.erainm.custominterceptor;

import com.erainm.customserializer.User;
import com.erainm.customserializer.UserSerializer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by erainm on 2021/9/10 8:47 上午.
 *
 * @Description TODO:
 **/
public class MyProducer {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        // 初始连接用到的broker地址
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "lg04:9092");
        // 指定key的序列化类
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        // 指定value的序列化类
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 如果有多个拦截器，则设置为多个拦截器的全限定类名，中间用逗号隔开
        configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
                "com.erainm.custominterceptor.InterceptorOne," +
                "com.erainm.custominterceptor.InterceptorTwo," +
                "com.erainm.custominterceptor.InterceptorThree");

        configs.put("classContent", "this is lg's kafka class");

        KafkaProducer<Integer, String> producer = new KafkaProducer<>(configs);

        ProducerRecord<Integer,String> record = new ProducerRecord<Integer,String>("tp_interceptor_01",0,1001,"this is lg's 1001 message");

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (null == exception) {
                    System.out.println(metadata.offset());
                }
            }
        });

        producer.close();
    }
}
